package com.bougaga.opensource.scheduler.redis.storage;

import com.bougaga.opensource.scheduler.core.task.DelayedTask;
import com.bougaga.opensource.scheduler.core.task.TaskStatus;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Redis任务存储实现
 * 
 * @author bougaga
 * @since 1.0.0
 */
@Component
public class RedisTaskStorage {
    
    private static final Logger logger = LoggerFactory.getLogger(RedisTaskStorage.class);
    
    private static final String TASK_KEY_PREFIX = "bou:scheduler:task:";
    private static final String TASK_INDEX_KEY = "bou:scheduler:task:index";
    private static final String TASK_STATUS_KEY = "bou:scheduler:task:status:";
    private static final String TASK_TYPE_KEY = "bou:scheduler:task:type:";
    private static final String TASK_EXECUTE_TIME_KEY = "bou:scheduler:task:execute_time";
    
    private final RedisTemplate<String, String> redisTemplate;
    private final ObjectMapper objectMapper;
    
    public RedisTaskStorage(RedisTemplate<String, String> redisTemplate, ObjectMapper objectMapper) {
        this.redisTemplate = redisTemplate;
        this.objectMapper = objectMapper;
    }
    
    /**
     * 保存任务
     */
    public boolean saveTask(DelayedTask task) {
        try {
            String taskJson = objectMapper.writeValueAsString(task);
            String taskKey = TASK_KEY_PREFIX + task.getTaskId();
            
            // 保存任务数据
            redisTemplate.opsForValue().set(taskKey, taskJson);
            
            // 设置过期时间（7天）
            redisTemplate.expire(taskKey, 7, TimeUnit.DAYS);
            
            // 添加到任务索引
            redisTemplate.opsForSet().add(TASK_INDEX_KEY, task.getTaskId());
            
            // 添加到状态索引
            redisTemplate.opsForSet().add(TASK_STATUS_KEY + task.getStatus().name(), task.getTaskId());
            
            // 添加到类型索引
            redisTemplate.opsForSet().add(TASK_TYPE_KEY + task.getTaskType(), task.getTaskId());
            
            // 添加到执行时间索引（ZSet，按执行时间排序）
            if (task.getExecuteTime() != null) {
                double score = task.getExecuteTime().toEpochSecond(ZoneOffset.UTC);
                redisTemplate.opsForZSet().add(TASK_EXECUTE_TIME_KEY, task.getTaskId(), score);
            }
            
            logger.debug("Task saved successfully: {}", task.getTaskId());
            return true;
        } catch (JsonProcessingException e) {
            logger.error("Failed to serialize task: {}", task.getTaskId(), e);
            return false;
        }
    }
    
    /**
     * 获取任务
     */
    public DelayedTask getTask(String taskId) {
        try {
            String taskKey = TASK_KEY_PREFIX + taskId;
            String taskJson = redisTemplate.opsForValue().get(taskKey);
            
            if (taskJson == null) {
                return null;
            }
            
            return objectMapper.readValue(taskJson, DelayedTask.class);
        } catch (JsonProcessingException e) {
            logger.error("Failed to deserialize task: {}", taskId, e);
            return null;
        }
    }
    
    /**
     * 更新任务
     */
    public boolean updateTask(DelayedTask task) {
        try {
            String taskJson = objectMapper.writeValueAsString(task);
            String taskKey = TASK_KEY_PREFIX + task.getTaskId();
            
            // 获取旧任务状态
            DelayedTask oldTask = getTask(task.getTaskId());
            if (oldTask != null) {
                // 从旧状态索引中移除
                redisTemplate.opsForSet().remove(TASK_STATUS_KEY + oldTask.getStatus().name(), task.getTaskId());
                
                // 从旧类型索引中移除
                redisTemplate.opsForSet().remove(TASK_TYPE_KEY + oldTask.getTaskType(), task.getTaskId());
                
                // 从执行时间索引中移除
                redisTemplate.opsForZSet().remove(TASK_EXECUTE_TIME_KEY, task.getTaskId());
            }
            
            // 保存新任务数据
            redisTemplate.opsForValue().set(taskKey, taskJson);
            
            // 添加到新状态索引
            redisTemplate.opsForSet().add(TASK_STATUS_KEY + task.getStatus().name(), task.getTaskId());
            
            // 添加到新类型索引
            redisTemplate.opsForSet().add(TASK_TYPE_KEY + task.getTaskType(), task.getTaskId());
            
            // 添加到执行时间索引
            if (task.getExecuteTime() != null) {
                double score = task.getExecuteTime().toEpochSecond(ZoneOffset.UTC);
                redisTemplate.opsForZSet().add(TASK_EXECUTE_TIME_KEY, task.getTaskId(), score);
            }
            
            logger.debug("Task updated successfully: {}", task.getTaskId());
            return true;
        } catch (JsonProcessingException e) {
            logger.error("Failed to serialize task: {}", task.getTaskId(), e);
            return false;
        }
    }
    
    /**
     * 删除任务
     */
    public boolean deleteTask(String taskId) {
        try {
            DelayedTask task = getTask(taskId);
            if (task == null) {
                return false;
            }
            
            String taskKey = TASK_KEY_PREFIX + taskId;
            
            // 删除任务数据
            redisTemplate.delete(taskKey);
            
            // 从索引中移除
            redisTemplate.opsForSet().remove(TASK_INDEX_KEY, taskId);
            redisTemplate.opsForSet().remove(TASK_STATUS_KEY + task.getStatus().name(), taskId);
            redisTemplate.opsForSet().remove(TASK_TYPE_KEY + task.getTaskType(), taskId);
            redisTemplate.opsForZSet().remove(TASK_EXECUTE_TIME_KEY, taskId);
            
            logger.debug("Task deleted successfully: {}", taskId);
            return true;
        } catch (Exception e) {
            logger.error("Failed to delete task: {}", taskId, e);
            return false;
        }
    }
    
    /**
     * 获取所有任务
     */
    public List<DelayedTask> getAllTasks() {
        try {
            Set<String> taskIds = redisTemplate.opsForSet().members(TASK_INDEX_KEY);
            if (taskIds == null || taskIds.isEmpty()) {
                return Collections.emptyList();
            }
            
            return taskIds.stream()
                    .map(this::getTask)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());
        } catch (Exception e) {
            logger.error("Failed to get all tasks", e);
            return Collections.emptyList();
        }
    }
    
    /**
     * 根据状态获取任务
     */
    public List<DelayedTask> getTasksByStatus(TaskStatus status) {
        try {
            Set<String> taskIds = redisTemplate.opsForSet().members(TASK_STATUS_KEY + status.name());
            if (taskIds == null || taskIds.isEmpty()) {
                return Collections.emptyList();
            }
            
            return taskIds.stream()
                    .map(this::getTask)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());
        } catch (Exception e) {
            logger.error("Failed to get tasks by status: {}", status, e);
            return Collections.emptyList();
        }
    }
    
    /**
     * 根据类型获取任务
     */
    public List<DelayedTask> getTasksByType(String taskType) {
        try {
            Set<String> taskIds = redisTemplate.opsForSet().members(TASK_TYPE_KEY + taskType);
            if (taskIds == null || taskIds.isEmpty()) {
                return Collections.emptyList();
            }
            
            return taskIds.stream()
                    .map(this::getTask)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());
        } catch (Exception e) {
            logger.error("Failed to get tasks by type: {}", taskType, e);
            return Collections.emptyList();
        }
    }
    
    /**
     * 获取待执行任务（按执行时间排序）
     */
    public List<DelayedTask> getPendingTasks() {
        try {
            Set<String> taskIds = redisTemplate.opsForZSet().range(TASK_EXECUTE_TIME_KEY, 0, -1);
            if (taskIds == null || taskIds.isEmpty()) {
                return Collections.emptyList();
            }
            
            return taskIds.stream()
                    .map(this::getTask)
                    .filter(Objects::nonNull)
                    .filter(task -> task.getStatus() == TaskStatus.PENDING)
                    .sorted(Comparator.comparing(DelayedTask::getExecuteTime))
                    .collect(Collectors.toList());
        } catch (Exception e) {
            logger.error("Failed to get pending tasks", e);
            return Collections.emptyList();
        }
    }
    
    /**
     * 获取到期的任务
     */
    public List<DelayedTask> getExpiredTasks() {
        try {
            long now = LocalDateTime.now().toEpochSecond(ZoneOffset.UTC);
            Set<String> taskIds = redisTemplate.opsForZSet().rangeByScore(TASK_EXECUTE_TIME_KEY, 0, now);
            
            if (taskIds == null || taskIds.isEmpty()) {
                return Collections.emptyList();
            }
            
            return taskIds.stream()
                    .map(this::getTask)
                    .filter(Objects::nonNull)
                    .filter(task -> task.getStatus() == TaskStatus.PENDING)
                    .collect(Collectors.toList());
        } catch (Exception e) {
            logger.error("Failed to get expired tasks", e);
            return Collections.emptyList();
        }
    }
    
    /**
     * 清理已完成任务
     */
    public int cleanupCompletedTasks(LocalDateTime beforeTime) {
        try {
            List<TaskStatus> completedStatuses = Arrays.asList(
                    TaskStatus.SUCCESS, TaskStatus.FAILED, TaskStatus.CANCELLED, TaskStatus.TIMEOUT
            );
            
            int cleanedCount = 0;
            for (TaskStatus status : completedStatuses) {
                List<DelayedTask> tasks = getTasksByStatus(status);
                for (DelayedTask task : tasks) {
                    if (task.getUpdateTime().isBefore(beforeTime)) {
                        if (deleteTask(task.getTaskId())) {
                            cleanedCount++;
                        }
                    }
                }
            }
            
            logger.info("Cleaned up {} completed tasks", cleanedCount);
            return cleanedCount;
        } catch (Exception e) {
            logger.error("Failed to cleanup completed tasks", e);
            return 0;
        }
    }
    
    /**
     * 获取任务数量
     */
    public long getTaskCount() {
        try {
            Long count = redisTemplate.opsForSet().size(TASK_INDEX_KEY);
            return count != null ? count : 0;
        } catch (Exception e) {
            logger.error("Failed to get task count", e);
            return 0;
        }
    }
    
    /**
     * 根据状态获取任务数量
     */
    public long getTaskCountByStatus(TaskStatus status) {
        try {
            Long count = redisTemplate.opsForSet().size(TASK_STATUS_KEY + status.name());
            return count != null ? count : 0;
        } catch (Exception e) {
            logger.error("Failed to get task count by status: {}", status, e);
            return 0;
        }
    }
}
